package com.tencent.iov.cowin.befsale.clue.service.clue.impl;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.software.boot.common.exception.BusinessException;
import com.tencent.iov.cowin.befsale.clue.config.properties.KafkaConsumerProperties;
import com.tencent.iov.cowin.befsale.clue.controller.clue.vo.response.ClueToMqResponse;
import com.tencent.iov.cowin.befsale.clue.dao.kafka.KafkaConfig;
import com.tencent.iov.cowin.befsale.clue.dao.kafka.KafkaProducer;
import com.tencent.iov.cowin.befsale.clue.dao.kafka.response.ClueInitialCrm;
import com.tencent.iov.cowin.befsale.clue.dao.kafka.response.ClueRecordSyncRequest;
import com.tencent.iov.cowin.befsale.clue.dao.kafka.response.ScrmLead;
import com.tencent.iov.cowin.befsale.clue.dao.redis.CommRedisDao;
import com.tencent.iov.cowin.befsale.clue.domain.clue.po.ClueLog;
import com.tencent.iov.cowin.befsale.clue.domain.cus.po.DealerCustomerFollowRecord;
import com.tencent.iov.cowin.befsale.clue.service.clue.ClueInitialService;
import com.tencent.iov.cowin.befsale.clue.service.clue.ClueLogService;
import com.tencent.iov.cowin.befsale.clue.service.clue.ClueService;
import com.tencent.iov.cowin.befsale.clue.service.clue.ClueSyncService;
import com.tencent.iov.cowin.befsale.clue.service.clue.KafkaService;
import com.tencent.iov.cowin.befsale.clue.service.cus.DealerCustomerFollowRecordService;
import com.tencent.iov.cowin.befsale.clue.value.contants.RedisKeyConst;
import com.tencent.iov.cowin.befsale.clue.value.enums.RecordSendMqStatusEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @Describe: kafka
 * @author: zly
 * @created: 2022-01-24
 * @version: 1.0
 */
@Service
@Slf4j
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    KafkaProducer producer;
    @Autowired
    ClueService clueService;
    @Autowired
    ClueInitialService clueInitialService;
    @Autowired
    DealerCustomerFollowRecordService dealerCustomerFollowRecordService;
    @Autowired
    CommRedisDao commRedisDao;
//    @Value("${scrm.receive:true}")
//    private Boolean receive;
//    @Value("${scrm.record.retry:3}")
//    private Integer recordRetry;
//
    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private KafkaConsumerProperties kafkaConsumerProperties;

    //同步老数据成功后：将crmClueId放入redis 由此键值触发record记录同步
    private String syncLeadTriggerRecord="LEAD:TRI:REC";
    /*@Autowired
    private ClueLogMapper clueLogMapper;*/

    @Autowired
    private ClueLogService clueLogService;
    @Autowired
    @Lazy
    private ClueSyncService clueSyncService;

//    @Autowired
//    private ObjectMapper objectMapper;

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 发送消息
     *
     * @param topic
     * @param obj
     */
    @Override
    public void send(String topic, Object obj) {
        try {
            producer.send(topic, obj);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消费kafka数据
     *
     * @param record
     * @throws JsonProcessingException
     */
//    @KafkaListener(topics = {KafkaConfig.customerSyncTopic}, groupId = KafkaConfig.customerSyncTopic_group_id)
    public void onCustomerToFellowSearchMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) throws IOException {
        Optional<?> value = Optional.ofNullable(record.value());
        if (!value.isPresent()) {
            return;
        }
        ObjectMapper objectMapper =new ObjectMapper();
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        ClueToMqResponse clueToMqResponse = objectMapper.readValue((String) value.get(), ClueToMqResponse.class);
        if (null == clueToMqResponse) {
            return;
        }

        Long count = commRedisDao.increment(RedisKeyConst.ALL_CLUE_TOMQ_KEY + clueToMqResponse.getUuId());

        if (count <= 1) {
            //修改记录的发送状态为已发送
            log.info("消费 线索记录 数据接收到的消息：{}", JSON.toJSONString(clueToMqResponse));
            DealerCustomerFollowRecord recordUpdate = new DealerCustomerFollowRecord();
            recordUpdate.setId(clueToMqResponse.getRId());
            recordUpdate.setSendMqStatus(RecordSendMqStatusEnum.SEND_YES.getCode());
            dealerCustomerFollowRecordService.update(recordUpdate);
        }
        ack.acknowledge();
    }

    /**
     * 消费sync-service服务发送的视图消息
     *
     * @param record
     * @throws JsonProcessingException
     */
//    @KafkaListener(topics = {KafkaConfig.syncToClueTopic}, groupId = KafkaConfig.syncToClueTopicGroupId)
    public void onSyncCLueMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) throws IOException {
        Optional<?> value = Optional.ofNullable(record.value());
        if (!value.isPresent()) {
            return;
        }
        ObjectMapper objectMapper =new ObjectMapper();
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        ScrmLead clueStatusRes = objectMapper.readValue((String) value.get(), ScrmLead.class);
        if (null == clueStatusRes) {
            return;
        }
        Long count = commRedisDao.increment(RedisKeyConst.ALL_CLUE_VIEW_KEY + clueStatusRes.getUuId());
        if (count <= 1) {
            log.info("消费视图信息 数据接收到的消息：clueStatusRes={}", JSON.toJSONString(clueStatusRes));
            clueService.sysnClueStatusFromMq(clueStatusRes);
        }
        ack.acknowledge();
    }


    /**
     * 消费sync-service 老数据线索同步
     *
     * @param record
     * @throws JsonProcessingException
     */
//    @KafkaListener(topics = {KafkaConfig.oldDataClue}, groupId = KafkaConfig.oldDataClueGroupId)
    public void onSyncOldLeadMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) throws IOException {
        Optional<?> value = Optional.ofNullable(record.value());
        if (!value.isPresent()) {
            return;
        }

        log.info("消费sync-service服务发送的线索数据记录={}", JSON.toJSONString(value.get()));
        if (!kafkaConsumerProperties.getReceive()) {
            ack.acknowledge();
            return;
        }

        long start = System.currentTimeMillis();
        ObjectMapper objectMapper =new ObjectMapper();
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        ClueInitialCrm clueLead = objectMapper.readValue((String) value.get(), ClueInitialCrm.class);
        if (null == clueLead) {
            return;
        }

        String key = RedisKeyConst.ALL_CLUE_OLD_DATA_KEY + clueLead.getCrmClueId();
        stringRedisTemplate.opsForValue().setIfAbsent(key, "0", 6, TimeUnit.HOURS);
        Long count = stringRedisTemplate.opsForValue().increment(key);

        if (count <= kafkaConsumerProperties.getRecordRetry()) {
            log.info("消费线索老数据信息 数据接收到的消息：clueLead={}", JSON.toJSONString(clueLead));
            //新增一条日志：
            Integer status = 0;
            ClueLog clueLog = new ClueLog();
            clueLog.setBody(JSON.toJSONString(clueLead));
            clueLog.setCrmClueId(clueLead.getCrmClueId());
            clueLog.setType(1);//线索同步1，记录同步2
            clueLog.setUuid(clueLead.getUuid());
            clueLog.setStatus(status);//'0成功，-1失败'
            clueLogService.insert(clueLog);

            String lockKey = String.format(RedisKeyConst.REDISSON_CLUE_KEY, clueLead.getCrmClueId());
            RLock rlock = redissonClient.getLock(lockKey);

            try {
                rlock.lock();
                clueInitialService.sysnOldClueFromMq(clueLead);
                status = 1;
            }catch (Exception e){
                e.printStackTrace();
                if(e instanceof BusinessException) {
                    clueLog.setErrorMessage(org.apache.commons.lang.StringUtils.substring(((BusinessException)e).getExtMessage(),0,200));
                }else {
                    clueLog.setErrorMessage(org.apache.commons.lang.StringUtils.substring(e.getMessage(), 0, 200));
                }
                status = -1;

            }finally {
                if(status == 1 || count.intValue() == kafkaConsumerProperties.getRecordRetry()){
                    ack.acknowledge();
                }
                clueLog.setStatus(status);
                clueLogService.updateById(clueLog);
                log.info("消费线索数据[{}]耗时：[{}]ms, 状态：[{}]", clueLead.getCrmClueId(), System.currentTimeMillis() - start, status);
                if (rlock.isLocked()) {
                    rlock.unlock();
                }
            }
        } else {
            ack.acknowledge();
        }

    }


    /**
     * 消费sync-service服务发送的线索数据跟进记录
     *
     * @param record
     * @throws IOException
     */
    @KafkaListener(topics = {KafkaConfig.oldDataClueTopic}, groupId = KafkaConfig.oldDataClueTopicGroupId)
    public void onOldClueRecordMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) throws IOException {
        Optional<?> value = Optional.ofNullable(record.value());
        if (!value.isPresent()) {
            return;
        }
        log.info("消费sync-service服务发送的线索数据跟进记录={}", JSON.toJSONString(value.get()));
        if (!kafkaConsumerProperties.getReceive()) {
            ack.acknowledge();
        } else {
            long start = System.currentTimeMillis();
            ObjectMapper objectMapper =new ObjectMapper();
            objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
            ClueRecordSyncRequest clueRecordRequest = objectMapper.readValue((String) value.get(), ClueRecordSyncRequest.class);
            if (null == clueRecordRequest) {
                return;
            }

            ClueLog clueLog = new ClueLog();
            clueLog.setBody(JSON.toJSONString(clueRecordRequest));
            clueLog.setType(2);
            clueLog.setCrmClueId(clueRecordRequest.getLeadId());
            clueLog.setUuid(clueRecordRequest.getUuid());
            clueLogService.insert(clueLog);

            Integer status = 0;
            String key = RedisKeyConst.ALL_CLUE_RECORD_KEY + clueRecordRequest.getLeadId();
            stringRedisTemplate.opsForValue().setIfAbsent(key, "0", 6, TimeUnit.HOURS);
            Long count = stringRedisTemplate.opsForValue().increment(key);

            if (count <= kafkaConsumerProperties.getRecordRetry()) {
                log.info("消费线索历史跟进记录数据：【{}】", JSON.toJSONString(clueRecordRequest));
                String lockKey = String.format(RedisKeyConst.REDISSON_CLUE_RECORD_KEY, clueRecordRequest.getLeadId());
                RLock rlock = redissonClient.getLock(lockKey);
                try {
                    rlock.lock();
                    clueSyncService.handleClueRecord(clueRecordRequest);
                    status = 1;
                } catch (BusinessException exception) {
                    exception.printStackTrace();
                    status = -1;
                    clueLog.setErrorMessage(StringUtils.substring(exception.getExtMessage(), 0, 200));
                    //本次消费掉，重新发送一条数据变更后的kafka
                    send(KafkaConfig.oldDataClueTopic,clueRecordRequest);
                } catch (RuntimeException e) {
                    e.printStackTrace();
                    status = -1;
                    clueLog.setErrorMessage(StringUtils.substring(e.getMessage(), 0, 200));
                    //本次消费掉，重新发送一条数据变更后的kafka
                    send(KafkaConfig.oldDataClueTopic,clueRecordRequest);
                } finally {
                    clueLog.setStatus(status);
                    clueLogService.updateById(clueLog);
                    ack.acknowledge();
                    log.info("消费线索跟进数据[{}]耗时：[{}]ms, 状态：[{}]", clueRecordRequest.getLeadId(), System.currentTimeMillis() - start, status);

                    if (rlock.isLocked()) {
                        rlock.unlock();
                    }
                }
            } else {
                ack.acknowledge();
            }
        }
    }


}
